package com.atguigu.app.dwd.db;

import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author shkstart
 * @creat 2022/9/3-16:28
 */
public class DwdInteractionComment {
    public static void main(String[] args) {
        //todo 1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //todo 2.设置检查点
        //todo 3.读取Kafka数据为表,过滤出评论表
        String topicName = "topic_db";
        String groupID = "dwd_interaction_review";
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "    `database` string,\n" +
                "    `table` string,\n" +
                "    `type` string,\n" +
                "    `ts` string,\n" +
                "    `data` map<String,String>,\n" +
                "    `old` map<String,String>,\n" +
                "    `pt` as proctime()\n" +
                ") " + KafkaUtil.getKafkaDDL(topicName,groupID));
//        tableEnv.sqlQuery("select * from topic_db").execute().print();
        //过滤出课程评论表
        Table commentInfo = tableEnv.sqlQuery("SELECT  `data`['id']           AS id,\n" +
                "        `data`['user_id']      AS user_id,\n" +
                "        `data`['course_id']    AS course_id,\n" +
                "        `data`['review_txt']   AS review_txt,\n" +
                "        `data`['review_stars'] AS review_stars,\n" +
                "        `data`['create_time']  AS create_time,\n" +
                "        `data`['deleted']      AS deleted,\n" +
                "        `ts`\n" +
                "FROM topic_db\n" +
                "WHERE `table` = 'review_info'\n" +
                "AND `type` = 'insert'");
        tableEnv.createTemporaryView("review_info",commentInfo);
//                tableEnv.sqlQuery("select * from comment_info").execute().print();
        //todo 6.建立Kafka输出表
        tableEnv.executeSql("CREATE TABLE kafka_result (" +
                "id string, \n" +
                "user_id string, \n" +
                "course_id string, \n" +
                "review_txt string, \n" +
                "review_stars string, \n" +
                "create_time string, \n" +
                "deleted string," +
                "ts string)" + KafkaUtil.getKafkaSinkDDL("dwd_interaction_review"));
        //todo 7.执行，查看结果
        tableEnv.executeSql("insert into kafka_result select * from review_info");
        tableEnv.sqlQuery("select * from kafka_result").execute().print();
    }
}
